package com.nl.networkflow;

import com.nl.bean.input.UserBehavior;
import java.net.URL;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;


/**
 * @author shihb
 * @date 2019/12/20 11:13
 * 每小时网站总浏览量(pv)的统计
 */
public class PageView {

  public static void main(String[] args) throws Exception {

    //1.创建执行环境
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.setParallelism(1);
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

    //2.source
    URL resource = PageView.class.getClassLoader().getResource("UserBehavior.csv");
    DataStreamSource<String> source = env.readTextFile(resource.getPath());

    //3.transform算子
    SingleOutputStreamOperator<Tuple2<String, Long>> dataStream = source
        .map(s->{
          String[] arr = s.split(",");
          long userId = Long.parseLong(arr[0].trim());
          long itemId = Long.parseLong(arr[1].trim());
          int categoryId = Integer.parseInt(arr[2].trim());
          String behavior = arr[3].trim();
          long timestamp = Long.parseLong(arr[4].trim());
          return UserBehavior.of(userId, itemId, categoryId, behavior, timestamp);
        })
        .assignTimestampsAndWatermarks(
            new BoundedOutOfOrdernessTimestampExtractor<UserBehavior>(Time.seconds(0)) {
              @Override
              public long extractTimestamp(UserBehavior element) {
                return element.getTimestamp() * 1000;
              }
            }
        )
        .filter(userBehavior -> "pv".equals(userBehavior.getBehavior()))
        .map(new MapFunction<UserBehavior, Tuple2<String, Long>>() {
          @Override
          public Tuple2 map(UserBehavior userBehavior) throws Exception {
            return new Tuple2<String, Long>("pv", 1L);
          }
        })
        //这边用lambda表达式会报错
//        .map(userBehavior -> new Tuple2<String, Long>("pv", 1L))
        .keyBy(0)
        // key的窗口函数 一小时的pv输出
        .timeWindow(Time.hours(1))
        .sum(1);

    dataStream.print("pv count:");

    env.execute("page view job");
  }

}
